package com.duowan.realtime.thrift;

import java.io.UnsupportedEncodingException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.duowan.realtime.thirft.api.DistinctData;
import com.duowan.realtime.uniq.algorithm.bloomfilter.BloomFilter;
import com.duowan.realtime.uniq.algorithm.bloomfilter.util.BloomFilterUtil;

public class ProxyBloomFilterServiceImpl {
	
	private static final String baseDir = "/data/backup/bloomfilter";
	
	private static final Logger LOG = LoggerFactory.getLogger("bflog");


	private BloomFilterDB bfDB;	
	
	public ProxyBloomFilterServiceImpl() {
		initBloomFilterDB();
	}

	private void initBloomFilterDB() {
		bfDB = new BloomFilterDB(baseDir);
	}

	private BloomFilter<String> chooseBloomFilter(String partition, String bloomfilterGroup) {
		return bfDB.getBloomFilterIfNotExistAndInit(bloomfilterGroup, partition);
	}



	public synchronized int notContainsCountAndAdd(String bloomfilterGroup, String partition, String group, List<String> values) {
		if (StringUtils.isBlank(bloomfilterGroup) || StringUtils.isBlank(partition) || StringUtils.isBlank(group)) {
			LOG.info("bloomfilter bloomfilterGroup or partition or group is blank," + " bloomfilterGroup=" + bloomfilterGroup + " parition=" + partition + " group="+ group);
		}
		BloomFilter<String> bloomFilter = chooseBloomFilter(partition, bloomfilterGroup);
		return BloomFilterUtil.notContainsCountAndAdd(group, values, bloomFilter);
	}
	

	public List<DistinctData> notContainsAndMark(String bloomfilterGroup, String partition, String group, List<DistinctData> distinctDatas) {
		if (StringUtils.isBlank(bloomfilterGroup) || StringUtils.isBlank(partition) || StringUtils.isBlank(group)) {
			LOG.info("bloomfilter bloomfilterGroup or partition or group is blank," + " bloomfilterGroup=" + bloomfilterGroup + " parition=" + partition + " group="+ group);
		}
		BloomFilter<String> bloomFilter = chooseBloomFilter(partition, bloomfilterGroup);
		return notContainsAndMark(group, distinctDatas, bloomFilter);
	}

	static List<DistinctData> notContainsAndMark(String group, List<DistinctData>  values, BloomFilter<String> bloomFilter) {
		
		int count = 0 ;
		for (DistinctData v : values) {
			if(StringUtils.isBlank(v.getDistinctKey()))  {
				continue;
			}
			
			byte[] bytes = null;
			try {
				bytes = (group + v).getBytes("UTF-8");
				if (bytes!=null && !bloomFilter.contains(bytes)) {
					count ++;
					v.setNotExist(true);
					bloomFilter.add(bytes);
				}
			} catch (UnsupportedEncodingException e) {
				throw new RuntimeException("error",e);
			}
		}
		LOG.debug("distinct counter group="  + group + ",source count=" + values.size() + ",distinct count=" + count );
		return values;
	}

}
